package com.myflink.day05;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

/**
 * @author Shelly An
 * @create 2020/9/22 9:44
 * 增量聚合：
 * 来一条处理一条，窗口关闭的时候，才会输出一次结果
 */
public class Window_IncrementalAgg {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> socketDS = env.socketTextStream("master", 4444);

        KeyedStream<Tuple2<String, Integer>, String> dataKS = socketDS.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        }).keyBy(r -> r.f0);

        //滚动窗口
        WindowedStream<Tuple2<String, Integer>, String, TimeWindow> dataWS = dataKS.timeWindow(Time.seconds(5));

        dataWS.reduce(
                //要求：两两数据类型一样，输出数据类型也一样
                new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                System.out.println(value1 +"<---->" +value2);
                return Tuple2.of(value1.f0, value1.f1+value2.f1);
            }
        }).print("reduce");

        dataWS.aggregate(
                //基本数据类型 不能用作泛型  int 包装类初始值是null，基本类型是0
                //输入数据的类型，  累加器的类型（中间聚合状态），  输出数据的类型
                new AggregateFunction<Tuple2<String, Integer>, Integer, Integer>() {
                    /**
                     * 创建累加器
                     * @return 初始值为0
                     */
                    @Override
                    public Integer createAccumulator() {
                        return 0;
                    }

                    /**
                     * 累加操作， 每来一条数据，如何进行累加
                     * @param value
                     * @param accumulator 计数操作
                     * @return 计数
                     */
                    @Override
                    public Integer add(Tuple2<String, Integer> value, Integer accumulator) {
                        System.out.println("add...");
                        return accumulator + 1;
                    }

                    /**
                     * 获取结果
                     * @param accumulator
                     * @return
                     */
                    @Override
                    public Integer getResult(Integer accumulator) {
                        System.out.println("result...");
                        return accumulator;
                    }

                    /**
                     * 会话窗口 才会调用，合并累加器的结果
                     * @param a
                     * @param b
                     * @return
                     */
                    @Override
                    public Integer merge(Integer a, Integer b) {
                        System.out.println("会话窗口调用merge");
                        return a+b;
                    }
                }).print("aggregate");

        env.execute();
    }
}
